package com.han.elasticsearch.kafka;

import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.han.elasticsearch.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author hanjl
 * @Description Kafka生产者
 * @ClassName KafkaProducer
 * @createTime 2021/12/22 16:39
 */
@Component
@Slf4j
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String,String> kafkaTemplate;

    public <T> void sendMessage(List<T> messages,String topic){
        if(CollectionUtils.isEmpty(messages)){
            log.error("topic {},message为空",topic);
            return;
        }
        if(StringUtils.isBlank(topic)){
            log.error("topic不可以为空");
        }
        log.info("准备向topic：{} 发送 {} 条数据",topic,messages.size());
        kafkaTemplate.send(topic, JsonUtil.toJsonString(messages));
        log.info("向topic：{} 发送 {} 条数据成功",topic,messages.size());
    }

    public <T> void sendMessageString(String messages,String topic){
        if(StringUtils.isBlank(messages)){
            log.error("topic {},message为空",topic);
            return;
        }
        if(StringUtils.isBlank(topic)){
            log.error("topic不可以为空");
        }
        log.info("准备向topic：{} 发送 {} 数据",topic,messages);
        kafkaTemplate.send(topic, JsonUtil.toJsonString(messages));
        log.info("向topic：{} 发送 {} 数据成功",topic,messages);
    }
}
